package com.jack.reactor;

import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;

/**
 * 使用Reactor库中的各种操作符
 * Created by zhang_j on 2019/10/23
 */
public class OperatorBufferTest {

    public static void main(String[] args) {
        bufferTest();
        filterTest();
        windowTest();
        zipWithTest();
        takeTest();
        reduceTest();
        mergeTest();
        flatMapTest();
        concatMapTest();
        combineLatestTest();
    }

    public static void combineLatestTest(){
        Flux.combineLatest(
                Arrays::toString,
                Flux.interval(Duration.of(100,ChronoUnit.MILLIS)).take(5),
                Flux.interval(Duration.of(50,ChronoUnit.MILLIS), Duration.of(100,ChronoUnit.MILLIS)).take(5)
        ).toStream().forEach(System.out::println);
    }

    public static void concatMapTest(){
        Flux.just(5,10)
                .concatMap(x->Flux.interval(Duration.of(x*10,ChronoUnit.MILLIS),Duration.of(100,ChronoUnit.MILLIS)).take(x))
                .toStream()
                .forEach(System.out::println);
    }

    public static void flatMapTest(){
        Flux.just(5,10)
                .flatMap(x->Flux.interval(Duration.of(x*10,ChronoUnit.MILLIS),Duration.of(100,ChronoUnit.MILLIS)).take(x))
                .toStream()
                .forEach(System.out::println);
        Flux.just(5,10)
                .flatMapSequential(x->Flux.interval(Duration.of(x*10,ChronoUnit.MILLIS),Duration.of(100,ChronoUnit.MILLIS)).take(x))
                .toStream()
                .forEach(System.out::println);
    }

    public static void mergeTest(){
        Flux.merge(Flux.interval(Duration.of(0,ChronoUnit.MILLIS), Duration.of(100,ChronoUnit.MILLIS))
                .take(5), Flux.interval(Duration.of(50,ChronoUnit.MILLIS), Duration.of(100,ChronoUnit.MILLIS))
                .take(5))
                .toStream()
                .forEach(System.out::println);
        Flux.mergeSequential(Flux.interval(Duration.of(0,ChronoUnit.MILLIS), Duration.of(100,ChronoUnit.MILLIS))
                .take(5), Flux.interval(Duration.of(50,ChronoUnit.MILLIS), Duration.of(100,ChronoUnit.MILLIS))
                .take(5))
                .toStream()
                .forEach(System.out::println);
    }

    public static void reduceTest(){
        Flux.range(1, 100).reduce((x, y) -> x + y).subscribe(System.out::println);
        Flux.range(1, 100).reduceWith(() -> 100, (x, y) -> x + y).subscribe(System.out::println);
    }

    public static void takeTest(){
        Flux.range(1, 1000).take(10).subscribe(System.out::println);
        Flux.range(1, 1000).takeLast(10).subscribe(System.out::println);
        Flux.range(1, 1000).takeWhile(i -> i < 10).subscribe(System.out::println);
        Flux.range(1, 1000).takeUntil(i -> i == 10).subscribe(System.out::println);
    }

    public static void zipWithTest(){
        Flux.just("a","b")
                .zipWith(Flux.just("c","d"))
                .subscribe(System.out::println);

        Flux.just("a","b")
                .zipWith(Flux.just("c","d"),(s1,s2)->String.format("%s-%s", s1, s2))
                .subscribe(System.out::println);
    }

    public static void windowTest(){
        Flux.range(1,100).window(20).subscribe(System.out::println);

        Flux.interval(Duration.of(100,ChronoUnit.MILLIS)).window(Duration.of(1001,ChronoUnit.MILLIS))
                .take(2).toStream().forEach(System.out::println);
    }

    public static void filterTest(){
        Flux.range(1, 10).filter(i -> i % 2 == 0).subscribe(System.out::println);


    }

    public static void bufferTest(){
        Flux.range(1,100).buffer(20).subscribe(System.out::println);

        Flux.interval(Duration.of(100, ChronoUnit.MILLIS)).buffer(Duration.of(1001, ChronoUnit.MILLIS))
                .take(2).toStream().forEach(System.out::println);

        Flux.range(1,10).bufferUntil(i->i%2==0).subscribe(System.out::println);

        Flux.range(1,10).bufferWhile(i->i%2==0).subscribe(System.out::println);
    }
}
